package State;

import bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.RichProcessWindowFunction;
import org.apache.flink.util.Collector;

/**
 * @author Spring_Hu
 * @date 2021/10/15 19:06
 */
public class Operator_state {
    public static void main(String[] args) throws Exception {
        //算子状态的定义
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(8000L);
        System.out.println(env.getCheckpointingMode().name());
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> source = env
                .socketTextStream("localhost", 9999)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] line = value.split(",");
                        return new WaterSensor(
                                line[0],
                                Long.parseLong(line[1]),
                                Integer.parseInt(line[2]));
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                    @Override
                    public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                        return element.getTs()*1000L;
                    }
                }));

        source.process(new MyProcess()).print();



        env.execute();
    }
    private static class MyProcess extends ProcessFunction<WaterSensor, String> implements CheckpointedFunction{
        String statevalue;
        ListState<String> listState;

        @Override
        //Checkpoint时会调用这个方法，我们要实现具体的snapshot逻辑，比如将哪些本地状态持久化
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            System.out.println("snapshotState......");

            listState.add(statevalue);

        }



        //初始化状态 向本地状态中填充数据. 每个子任务调用一次
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            System.out.println("initializeState......");
            listState = context.getOperatorStateStore().getListState(new ListStateDescriptor("list_state", Types.STRING));
            for (String s : listState.get()) {
                System.out.println("拿到的状态值："+s);
            }
        }

        @Override
        public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
            System.out.println("processElement....");
            statevalue=value.toString();
            out.collect(value.toString());
        }
    }



}
